package org.apache.ocean.solr.replication;

import java.io.EOFException;
import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.EnumSet;
import java.util.List;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;

import org.apache.commons.lang.builder.ReflectionToStringBuilder;
import org.apache.ocean.main.fields.SchemaField.Attribute;
import org.apache.ocean.main.replication.ByteBufferPool.ByteBuffer;
import org.apache.ocean.main.replication.ByteBufferPool.ByteBuffers;
import org.apache.ocean.main.util.ByteArrayOutputStream;

// TODO: add footer to insure record is complete
// TODO: add position of previous header
// TODO: add to recordheader if documents portion is ramdirectory
public class RawLogFile {
	public static Logger log = Logger.getLogger(RawLogFile.class.getName());
	public static final byte DELETED = 101;
	public static final byte OK = 9;
	public static final byte[] HEADER = new byte[] { 'O', 'c', 'N' };

	public static enum Load {
		DOCUMENTS, DELETES
	};

	private EnumSet<Attribute> attributes = EnumSet.noneOf(Attribute.class);
	public static final int DATA_START = 256;
	public static final int HEADER_LENGTH = 192;
	private RandomAccessFile writeRandomAccessFile;
	private File file;
	private long currentWritePosition = DATA_START;
	private ReentrantLock writeLock = new ReentrantLock();

	public RawLogFile(File file) throws IOException {
		this.file = file;
		writeRandomAccessFile = new RandomAccessFile(file, "rw");
	}

	public long size() {
		return file.length();
	}

	public static class Data {
		private ByteBuffers documentByteBuffers;
		private ByteBuffers deletesByteBuffers;

		public Data(ByteBuffers documentByteBuffers, ByteBuffers deletesByteBuffers) {
			this.documentByteBuffers = documentByteBuffers;
			this.deletesByteBuffers = deletesByteBuffers;
		}

		public void finished() {
			if (documentByteBuffers != null) {
				documentByteBuffers.finished();
			}
			if (deletesByteBuffers != null) {
				deletesByteBuffers.finished();
			}
		}

		public ByteBuffers getDocumentsByteBuffers() {
			return documentByteBuffers;
		}

		public ByteBuffers getDeletesByteBuffers() {
			return deletesByteBuffers;
		}
	}

	public Data readData(EnumSet<Load> load, RecordHeader recordHeader, ByteBufferPool documentBytePool, ByteBufferPool deletesBytePool) throws IOException {
		RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r");
		try {
			ByteBuffers documentsBuffers = null;
			ByteBuffers deletesBuffers = null;
			if (load.contains(Load.DOCUMENTS)) {
				// ByteBuffer documentsBuffer =
				// documentBytePool.get(recordHeader.documentsDataLength);
				//int dataLength = recordHeader.documentsDataLength;
				randomAccessFile.seek(recordHeader.documentsDataStart);
				//ByteBuffer byteBuffer = documentBytePool.get(1024 * 16);
				//randomAccessFile.read(byteBuffer.getBytes(), 0, len);
				// byte[] bytes = documentsBuffer.getBytes();
				// randomAccessFile.read(bytes, 0, recordHeader.documentsDataLength);
				documentsBuffers = new ByteBuffers(randomAccessFile, 1024 * 32, recordHeader.documentsDataLength, documentBytePool);
			}
			if (load.contains(Load.DELETES)) {
				//ByteBuffer deletesBuffer = deletesBytePool.get(recordHeader.deletesDataLength);
				randomAccessFile.seek(recordHeader.deletesDataStart);
				deletesBuffers = new ByteBuffers(randomAccessFile, 1024 * 4, recordHeader.deletesDataLength, deletesBytePool);
				//byte[] bytes = deletesBuffer.getBytes();
				//randomAccessFile.read(bytes, 0, recordHeader.deletesDataLength);
				//deletesBuffers = new ByteBuffers(deletesBuffer, recordHeader.deletesDataLength);
			}
			return new Data(documentsBuffers, deletesBuffers);
		} finally {
			if (randomAccessFile != null)
				randomAccessFile.close();
		}
	}

	public List<RecordHeader> loadRecordHeaders() throws IOException {
		LoadRecordHeaders loadRecordHeaders = new LoadRecordHeaders();
		return loadRecordHeaders.getRecordHeaders();
	}

	public class LoadRecordHeaders {
		private long currentReadPosition = DATA_START;
		private List<RecordHeader> recordHeaders = new ArrayList<RecordHeader>();

		public LoadRecordHeaders() throws IOException {
			RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r");
			try {
				while (true) {
					randomAccessFile.seek(currentReadPosition);
					RecordHeader recordHeader = new RecordHeader();
					byte[] header = new byte[3];
					randomAccessFile.readFully(header);
					if (!Arrays.equals(HEADER, header)) {
						// TODO: scan to next header if there is one
						log.severe("log file corrupted " + file.getAbsolutePath());
						while (true) {
							randomAccessFile.readFully(header);
							if (Arrays.equals(HEADER, header)) {
								break;
							}
						}
						//continue;
					}
					byte status = randomAccessFile.readByte();
					int nextHeaderPosition = randomAccessFile.readInt();
					if (status == DELETED) {
						currentReadPosition = nextHeaderPosition;
						continue;
					}
					recordHeader.id = randomAccessFile.readLong();
					recordHeader.headerStart = randomAccessFile.readInt();
					recordHeader.documentsDataStart = randomAccessFile.readInt();
					recordHeader.documentsDataLength = randomAccessFile.readInt();
					recordHeader.deletesDataStart = randomAccessFile.readInt();
					recordHeader.deletesDataLength = randomAccessFile.readInt();
					recordHeader.category = randomAccessFile.readUTF();
					recordHeader.schemaVersion = randomAccessFile.readInt();
					recordHeaders.add(recordHeader);
					currentReadPosition = recordHeader.headerStart + HEADER_LENGTH + recordHeader.documentsDataLength + recordHeader.deletesDataLength;
				}
			} catch (EOFException eofException) {
				// at end of file
			} finally {
				if (randomAccessFile != null)
					randomAccessFile.close();
			}
		}

		public List<RecordHeader> getRecordHeaders() {
			return recordHeaders;
		}
	}

	public boolean delete(RecordHeader recordHeader) throws IOException {
		long lastPosition = writeRandomAccessFile.getFilePointer();
		try {
			writeLock.lock();
			try {
				writeRandomAccessFile.seek(recordHeader.headerStart);
				byte[] header = new byte[3];
				writeRandomAccessFile.readFully(header);
				if (!Arrays.equals(HEADER, header)) {
					throw new IOException("no header");
				}
				byte first = (byte) writeRandomAccessFile.read();
				if (first == OK) {
					int nextHeaderPosition = writeRandomAccessFile.readInt();
					int idBytesLength = writeRandomAccessFile.readInt();
					byte[] idBytes = new byte[idBytesLength];
					writeRandomAccessFile.readFully(idBytes);
					BigInteger id = new BigInteger(idBytes);
					assert id.equals(recordHeader.id);
					writeRandomAccessFile.seek(recordHeader.headerStart + 4);
					writeRandomAccessFile.write(DELETED);
					writeRandomAccessFile.getFD().sync();
					return true;
				} else {
					// already deleted
					return false;
				}
			} finally {
				writeRandomAccessFile.seek(lastPosition);
			}
		} finally {
			writeLock.unlock();
		}
	}

	// TODO: add read to footer to insure the record was fully written
	// TODO: handle write failure cleanly
	public RecordHeader write(Long id, String category, long schemaVersion, Data data) throws IOException {
		writeLock.lock();
		try {
			ByteBuffers documentsByteBuffers = data.getDocumentsByteBuffers();
			ByteBuffers deletesByteBuffers = data.getDeletesByteBuffers();
			RecordHeader recordHeader = new RecordHeader();
			recordHeader.id = id;
			recordHeader.headerStart = (int) currentWritePosition;
			recordHeader.documentsDataStart = (int) currentWritePosition + HEADER_LENGTH;
			recordHeader.documentsDataLength = documentsByteBuffers.getLength();
			recordHeader.deletesDataStart = recordHeader.documentsDataStart + recordHeader.documentsDataLength;
			recordHeader.deletesDataLength = deletesByteBuffers.getLength();
			recordHeader.category = category;
			recordHeader.schemaVersion = (int) schemaVersion;
			int nextHeaderPosition = HEADER_LENGTH + documentsByteBuffers.getLength() + deletesByteBuffers.getLength();
			writeRandomAccessFile.setLength(writeRandomAccessFile.length() + HEADER_LENGTH + documentsByteBuffers.getLength() + deletesByteBuffers.getLength());
			writeRandomAccessFile.seek(recordHeader.headerStart);
			writeRandomAccessFile.write(HEADER);
			writeRandomAccessFile.write(OK);
			writeRandomAccessFile.writeInt(nextHeaderPosition);
			writeRandomAccessFile.writeLong(id);
			writeRandomAccessFile.writeInt(recordHeader.headerStart);
			writeRandomAccessFile.writeInt(recordHeader.documentsDataStart);
			writeRandomAccessFile.writeInt(recordHeader.documentsDataLength);
			writeRandomAccessFile.writeInt(recordHeader.deletesDataStart);
			writeRandomAccessFile.writeInt(recordHeader.deletesDataLength);
			writeRandomAccessFile.writeUTF(recordHeader.category);
			writeRandomAccessFile.writeInt(recordHeader.schemaVersion);
			writeRandomAccessFile.seek(recordHeader.documentsDataStart);
			documentsByteBuffers.writeTo(writeRandomAccessFile);
			deletesByteBuffers.writeTo(writeRandomAccessFile);
			currentWritePosition = HEADER_LENGTH + recordHeader.headerStart + documentsByteBuffers.getLength() + deletesByteBuffers.getLength();
			writeRandomAccessFile.getFD().sync();
			return recordHeader;
		} finally {
			writeLock.unlock();
		}
	}
  
	public void close() throws IOException {
		writeRandomAccessFile.close();
	}

	public static class RecordHeader {
		public long id;
		public int headerStart;
		public int documentsDataStart;
		public int documentsDataLength;
		public int deletesDataStart;
		public int deletesDataLength;
		public String category;
		public int schemaVersion;

		public String toString() {
			return ReflectionToStringBuilder.toString(this);
		}
	}
}
